package org.company.saturn.hadoop.yarn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.company.saturn.hadoop.yarn.map.CounterMapper;

/**
 * 计数器的Main入口类
 * @author hucw
 * 响应时间	IP	请求类型	响应状态
 * 23	192.168.10.123	GET		200
 * 24	192.168.10.124			404
 * 25	192.168.10.125	GET		505
 * 23	192.168.10.123	GET		200
 * 24	192.168.10.124			404
 * 25	192.168.10.125	GET		505
 *
 * hadoop job -counter [-counter]
 * hadoop job -counter "ERROR_LOG_TIME"
 */
public class CounterMain extends Configured implements Tool {

	@Override
	public int run(String[] args) throws Exception {
		Configuration conf = this.getConf();
		String[] otherArgs = new GenericOptionsParser(args).getRemainingArgs();
		if(otherArgs.length != 2) {
			System.err.println("Usage: MyCounter <in> <out>");
			return 2;
		}
		Job job = Job.getInstance(conf,"MyCounter");
		job.setJarByClass(CounterMain.class);
		job.setMapperClass(CounterMapper.class);
		
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new CounterMain(), args);
		System.out.println(exitCode);
	}
}
